package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3CreateDataFream {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .appName("spark")
      .master("local")
      .getOrCreate()


    import spark.implicits._


    /**
      *
      * 创建DF的方式
      * 1、读取JSON格式的数据创建DF
      * 2、读取CSV格式的数据创建DF
      * 3、通过JDBC创建DF
      * 4、可以将RDD转换成DF
      *
      */

    //读取JSON格式的数据创建DF
    //读取json格式不需要指定列名，json中自带列名
    val jsonDF: DataFrame = spark
      .read
      .format("json")
      .load("spark/data/students.json")

    jsonDF.show()


    //、读取CSV格式的数据创建DF
    val csvDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //数据分割方式，默认逗号分割
      //.option("header", "true") //第一行作为列名
      .schema("id STRING, name STRING , age INT , gender STRING ,clazz STRING") //指定表结构
      .load("spark/data/students.txt")

    csvDF.show()





    //4、可以将RDD转换成DF

    //通过sparkSession 获取SparkContext
    val sc: SparkContext = spark.sparkContext


    val rdd: RDD[String] = sc.textFile("spark/data/students.txt")


    val rdd2: RDD[(String, String, Int, String, String)] = rdd.map(line => {
      val split: Array[String] = line.split(",")

      (split(0), split(1), split(2).toInt, split(3), split(4))
    })


    /**
      * 将rddz转换成DF的时候加上列名
      *
      */
    val rddToDF: DataFrame = rdd2.toDF("id", "name", "age", "gender", "clazz")

    rddToDF.show()



    //通过JDBC创建DF

    val jdbcDF: DataFrame = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "student.student")
      .option("user", "root")
      .option("password", "123456")
      .load()

    jdbcDF.show()


  }

}
